package com.sky.kafka.consumer.web;

import com.sky.kafka.consumer.util.MQDict;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * 拉取kafka消息
 */
@RestController
@RequestMapping("consumer")
public class ConsumerController {

    final int minBatchSize = 5;

    private static KafkaConsumer<String,String> consumer;

    /**
     *  初始化消费者
     */
    static {
        Properties configs = initConfig();
        consumer = new KafkaConsumer<String, String>(configs);
        // 订阅主题 不使用下面设置，将会报：Consumer is not subscribed to any topics or assigned any partitions
        consumer.subscribe(Arrays.asList(MQDict.CONSUMER_TOPIC));
    }
    /**
     *  初始化配置
     */
    private static Properties initConfig(){
        Properties props = new Properties();
        props.put("bootstrap.servers", MQDict.MQ_ADDRESS_COLLECTION);
        props.put("group.id", MQDict.CONSUMER_GROUP_ID);
        props.put("enable.auto.commit", MQDict.CONSUMER_ENABLE_AUTO_COMMIT);
        props.put("auto.commit.interval.ms", MQDict.CONSUMER_AUTO_COMMIT_INTERVAL_MS);
        props.put("session.timeout.ms", MQDict.CONSUMER_SESSION_TIMEOUT_MS);
        props.put("max.poll.records", MQDict.CONSUMER_MAX_POLL_RECORDS);
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        return props;
    }

    @GetMapping("poll")
    public void consumer(){
        List<ConsumerRecord<String,String>> buffer = new ArrayList<>(minBatchSize);
        try {
            while (true) {
                // 参数中的超时时间根据业务逻辑来设置，尽量稍微长一点，避免获取不到消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));
                if (records.isEmpty()) {
                    consumer.wakeup();
                }
                // 给list中add 值
                records.forEach(buffer::add);
                if (buffer.size() >= minBatchSize) {
                    records.forEach(record -> {
                        System.out.println("topic = " + record.topic() + ", partition = " + record.partition() +
                                ", offset = " + record.offset() + ", customer = " + record.key() + ", country = " + record.value());

                    });
                    // 业务处理完成之后，同步提交位移
                    consumer.commitSync();
                }
            }
        }catch (Exception e) {}
    }
}
